package dataflowscheme.engine.core;

import java.util.Calendar;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.log4j.Logger;

/**
 * Event Parser thread is processing the Event queue - in each 
 * iteration it takes one event (or thread waits until an event 
 * is available) and invoke its processing dependent on type of an event .
 * 
 * @author <a href="mailto:misiak7@gmail.com">Michal Antolik</a>
 *
 */
public class EventParser implements Runnable{

	static Logger log = Logger.getLogger(EventParser.class);
	
	/** queue where are events of type <code>IEvent<code> put  */
	private final LinkedBlockingQueue<IEvent> queue;
	
	/** reference to activity checker to announce current state */
	private final ActivityChecker activityChecker;
	
	/** when true, then parser will end processing */
	private AtomicBoolean stop = new AtomicBoolean(false);
	
	/** if statistics should be recorded */
	private final boolean printStats;
	
	/** instance of class which process statistics */
	private final ExecutionEPstats stats = ExecutionEPstats.getInstance();;
	
	public EventParser(LinkedBlockingQueue<IEvent> queue, ActivityChecker activityChecker, boolean prinStats) {
		this.queue = queue;
		this.activityChecker = activityChecker;
		this.printStats = prinStats;
	}
	
	
	public void run() {
		try{
			while(!stop.get()){
				if(printStats){
					processEventsWithStats();
				} else {
					processEventsWithoutStats();
				}
			}
			activityChecker.setEventParserInactivity(); //parser is going to be shutdown
			if(queue.size() != 0){
				log.warn("Ending - remaining unprocessed events: " + queue.size());
			}
		}catch(InterruptedException ie){
			log.debug("Event parser interrupted", ie);
		} finally {
			log.debug("I'm leaving, arrivederci!");
		}
	}
	
	/**
	 * stop the parser - it is called from the end event
	 */
	public void stop(){
		stop.set(true);
	}
	
	private void processEventsWithStats() throws InterruptedException{
			long beforeTake = Calendar.getInstance().getTimeInMillis();
		activityChecker.setEventParserInactivity(); //thread can be blocked by take() operation
		
		IEvent e = queue.take();
		
		activityChecker.setEventParserActivity();
			long afterTake = Calendar.getInstance().getTimeInMillis();
			long waitingTime = afterTake - beforeTake;
		
		e.execute();
		
			long serviceTime = Calendar.getInstance().getTimeInMillis() - afterTake;
		
			//submit statistics
			stats.submitEventParserTime(serviceTime, waitingTime);
	}
	
	private void processEventsWithoutStats() throws InterruptedException{
		activityChecker.setEventParserInactivity(); //thread can be blocked by take() operation
		IEvent e = queue.take();
		activityChecker.setEventParserActivity();

		e.execute();
	}

}
